#include <sys/socket.h>
#include "comet_connection.h"

#include <boost/bind.hpp>
#include <boost/typeof/typeof.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/thread.hpp>
#include <boost/make_shared.hpp>

#include "base/logging.h"
#include "base/url_encode.h"
#include "base/json_escape.h"
#include "base/json_parser.h"
#include "uid_comet_conn_map.h"

using namespace std;

namespace imserver {

CometConnection::CometConnection(boost::asio::io_service& io_service)
  : io_service_(io_service)
  , socket_(io_service)
  , read_buf_begin_(0)
  , read_buf_end_(0)
  , write_buf_(NULL)
  , write_buf_begin_(0)
  , write_buf_end_(0)
  , uid_(0)
  , ready_(false)
  , valid_(false)
{
}

CometConnection::~CometConnection() {
}

CometConnectionPtr CometConnection::Create(boost::asio::io_service& io_service) {
  return CometConnectionPtr(new CometConnection(io_service));
}

bool CometConnection::isHttpHeaderCompleted(const char * data, size_t len) {
  return std::string::npos != std::string(data, len).find("\r\n\r\n");
}

bool CometConnection::isHttpBodyCompleted(const std::string method,
    const std::map<std::string, std::string>& http_headers,
    const std::string& http_body) {
  if (method != "POST") {
    return true;
  }
  long content_length = 0L;
  BOOST_AUTO(it, http_headers.find("Content-Length"));
  if (it != http_headers.end()) {
    try {
      content_length = boost::lexical_cast<size_t>(it->second);
    } catch (boost::bad_lexical_cast &) {}
  }
  if (http_body.size() < content_length) {
    return false;
  }
  return true;
}

bool CometConnection::parseKeyValue(const char * data, const char * key_seps,
    char kv_sep, std::map<std::string, std::string> * pairs) {
  if (!data)
    return false;

  std::vector<std::string> strs;
  boost::split(strs, data, boost::is_any_of(key_seps), boost::token_compress_on);
  for (size_t i = 0; i < strs.size(); ++i) {
    size_t pos = strs[i].find_first_of(kv_sep);
    if (pos != std::string::npos) {
      (*pairs)[strs[i].substr(0, pos)] = base::UrlDecode(strs[i].substr(pos + 1));
    } else {
      (*pairs)[strs[i]] = "";
    }
  }
  return true;
}

void CometConnection::Start() {
  LOG_INFO("CometConnection::Start --> begin");
  ip::tcp::no_delay nodelay(true);
  socket_.set_option(nodelay);
  
  boost::asio::socket_base::linger linger(true, 0);
  socket_.set_option(linger);

  boost::asio::socket_base::keep_alive keep_alive(true);
  socket_.set_option(keep_alive);

  int fd = socket_.native_handle();
  int optval = 5;
  setsockopt(fd, SOL_TCP, TCP_KEEPCNT, &optval, sizeof(optval));
  optval = 20;
  setsockopt(fd, SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(optval));
  optval = 5;
  setsockopt(fd, SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(optval));
  
  boost::asio::socket_base::send_buffer_size send_buf_size(1024);
  socket_.set_option(send_buf_size);

  valid_ = true;
  AsyncRead();
}

std::string CometConnection::remote_addr() const {
  boost::system::error_code ec;
  ip::tcp::endpoint endpoint = socket_.remote_endpoint(ec);
  if (ec) {
    // An error occurred.
    return "bad_addr";
  }
  return endpoint.address().to_string();
}


void CometConnection::Abort() {
  /*if (http_request_) {
    http_request_.reset();
  }*/

  closeSocket();
  return;
}

void CometConnection::AsyncRead() {
  LOG_INFO("CometConnection::AsyncRead --> begin");
  socket_.async_read_some(boost::asio::buffer(read_buf_ + read_buf_end_, kReadBufLength - read_buf_end_),
      boost::bind(&CometConnection::HandleRead, shared_from_this(),
    boost::asio::placeholders::error, // 占位符
    boost::asio::placeholders::bytes_transferred));
}

void CometConnection::AsyncWrite() {
  LOG_INFO("CometConnection::AsyncWrite --> begin");
  if (socket_.is_open()) {
    boost::asio::async_write(socket_,
        boost::asio::buffer(write_buf_ + write_buf_begin_, write_buf_end_ - write_buf_begin_),
        boost::bind(&CometConnection::HandleWrite, shared_from_this(),
            boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
  } else {
    LOG_ERROR("CometConnection::AsyncWrite --> try to write on bad socket. uid:" << uid_);
    valid_ = false;
    ready_ = false;
    UidCometConnMap::instance().remove(uid_);
    /*if (http_request_) {
      http_request_->OnWriteError();
      http_request_.reset();
    }*/
  }
}

void CometConnection::HandleWrite(const boost::system::error_code& error,
    size_t bytes_transferred) {
  if (error) {
    LOG_ERROR("CometConnection::HandleWrite --> error:" << error << " errorMsg:" << error.message() << " uid:" << uid_);
    /*if (http_request_) {
      http_request_->OnWriteError();
      http_request_.reset();
    }*/

    closeSocket();
    UidCometConnMap::instance().remove(uid_);
    return;
  }

  /*if (!http_request_) {
    // socket_.close(); // 不需要close
    return;
  }*/

  write_buf_begin_ += bytes_transferred;
  if (write_buf_begin_ < write_buf_end_) {
    AsyncWrite();
  } else {
    /*http_request_->OnWriteComplete();
    http_request_.reset();*/
    // socket_.close();
    std::string msg(write_buf_);
    onWriteComplete(msg);
  }
}

/*HttpRequest * CometConnection::CreateRequest() {
  std::string data(read_buf_, read_buf_end_);
  size_t url_begin = data.find_first_of(' ');
  if (url_begin == std::string::npos) {
    return NULL;
  }
  ++url_begin;
  size_t url_end = data.find_first_of(' ', url_begin);
  if (url_end == std::string::npos) {
    return NULL;
  }
  std::string url = data.substr(url_begin, url_end - url_begin); 
  
  size_t pos = url.find_first_of("#?");
  std::string path(url);
  if (pos != std::string::npos) {
    path = url.substr(0, pos);
  }

  if (path == "/subscribe") {
    return new CometSubscribeRequest(shared_from_this(), data);
  }

  if (path == "/favicon.ico") {
    return new StaticFileRequest(shared_from_this(), data, "./favicon.ico");
  }

  if (path == "/chat-channel.html") {
    return new StaticFileRequest(shared_from_this(), data, "./chat-channel.html");
  }

  if (path == "/ajax-proxy.html") {
    return new StaticFileRequest(shared_from_this(), data, "./ajax-proxy.html");
  }

  if (path == "/publish") {
    return new CometPublishRequest(shared_from_this(), data);
  }

  if (path == "/admin/stat.txt") {
    return new StatRequest(shared_from_this(), data);
  }

  if (path == "/debug.html") {
    return new StaticFileRequest(shared_from_this(), data, "./debug.html");
  }

  return new DefaultHttpRequest(shared_from_this(), data);
}*/

void CometConnection::HandleRead(const boost::system::error_code& error, size_t bytes_transferred) {
  LOG_INFO("CometConnection::HandleRead --> begin");
  if (error) {
    if (error != boost::asio::error::eof) {
      LOG_ERROR("CometConnection::HandleRead --> error:" << error << " errorMsg:" << error.message() << " uid_:" << uid_);
    }

    /*if (http_request_) {
      // http_request_ 的 socket 一律在request对象dtor中进行关闭,
      // 原因是，除了网络事件导致socket关闭，duplicate请求也会导致关闭
      http_request_->OnReadError();
      http_request_.reset();
    }*/
    closeSocket();
    UidCometConnMap::instance().remove(uid_);
    return;
  }
  read_buf_end_ += bytes_transferred;

  /*if (read_buf_end_ >= kReadBufLength) {
    if (http_request_) {
      http_request_->OnReadError();
      http_request_.reset();
    }
    socket_.close();
    return;
  }*/

  if (isHttpHeaderCompleted(read_buf_, read_buf_end_)) {
    /*if (http_request_) {*/
    if (!http_headers_.empty()) {
      /*http_request_->AppendBody(read_buf_ + read_buf_begin_, read_buf_end_ - read_buf_begin_);*/
      http_body_.append(read_buf_ + read_buf_begin_, read_buf_end_ - read_buf_begin_);
    } else {
      LOG_INFO("CometConnection::HandleRead --> http header completed. uid_:" << uid_);
      /*HttpRequest * req = CreateRequest();
      if (req == NULL) {
        LOG_WARN("CreateRequest fail!");
        socket_.close();
        return;
      }
      http_request_.reset(req);*/
      if (!buildHttpRequest(std::string(read_buf_, read_buf_end_))) {
        closeSocket();
        UidCometConnMap::instance().remove(uid_);
        return;
      }
    }

    read_buf_begin_ += read_buf_end_;
    /*if (!http_request_->IsLegal()) {
      LOG_WARN("Bad(non-HTTP) Request addr=" << remote_addr());
      http_request_.reset();
      // socket_.close();
      return;
    }*/

    if (isHttpBodyCompleted("POST", http_headers_, http_body_)) {
      LOG_INFO("CometConnection::HandleRead --> http body completed. uid_:" << uid_);
      handleHttpRequest();
      ready_ = true;
      UidCometConnMap::instance().add(uid_, shared_from_this());
    }
  }

  /*if (http_request_ && http_request_->IsBodyCompleted()) {
    http_request_->OnReadComplete();
  }*/
  AsyncRead(); // 即使数据读完，也要继续读, 以检测网络错误
}

bool CometConnection::buildHttpRequest(const std::string & data) {
  size_t end_of_header = data.find("\r\n\r\n");
  if (end_of_header == std::string::npos) {
    LOG_ERROR("CometConnection::buildHttpRequest --> invalid header. data: " << data);
    return false;
  }

  size_t pos = data.find("\r\n");
  std::string request_line = data.substr(0, pos);
  std::vector<std::string> fields;
  boost::split(fields, request_line, boost::is_any_of(" "), boost::token_compress_on);
  if (fields.size() != 3) {
    LOG_ERROR("CometConnection::buildHttpRequest --> request data is not 3 line. data:" << data);
    return false;
  }

  std::string method = fields[0];
  if (method != "POST") {
    LOG_ERROR("CometConnection::buildHttpRequest --> request method is not POST. data:" << data);
    return false;
  }
  size_t url_end_pos = fields[1].find_first_of('?');
  http_url_ = fields[1].substr(0, url_end_pos);

  while(pos < end_of_header - 4) {
    size_t next_pos = data.find("\r\n", pos + 2);
    if (next_pos == std::string::npos) {
      break;
    }
    std::string line = data.substr(pos + 2, next_pos - pos - 2);
    size_t sp = line.find(": ");
    if (sp != std::string::npos && sp > 0 && sp < line.size() - 2) {
      http_headers_[line.substr(0, sp)] = line.substr(sp + 2);
    }
    pos = next_pos;
  }

  std::map<std::string,std::string>::const_iterator it = http_headers_.find("Cookie");
  if (it != http_headers_.end()) {
    parseKeyValue(it->second.c_str(), "; ", '=', &http_cookies_);
  }
  // 从Cookie中取出uid和ticket，调用验票服务验票
  // uid_ =
  uid_ = 1;

  http_body_ = data.substr(end_of_header + 4);

  return true;
}

void CometConnection::handleHttpRequest() {
  if (!http_body_.empty()) {
    parseKeyValue(http_body_.c_str(), "&", '=', &http_post_params_);
  }
}

void CometConnection::onWriteComplete(const std::string& msg) {
  std::map<std::string, std::string> msg_content_map;
  size_t body_start = msg.find("\r\n\r\n");
  if (body_start == std::string::npos) {
    LOG_ERROR("CometConnection::HandleWrite --> can not find \r\n\r\n. msg:" << msg);
    return;
  }
  body_start += 4;
  std::string msg_t = msg.substr(body_start);
  base::JsonParser::instance().parse(msg_t, msg_content_map);
  if (msg_content_map.find("id") == msg_content_map.end() || msg_content_map["id"].empty()) {
    LOG_ERROR("CometConnection::HandleWrite --> can not find id. msg_t:" << msg_t);
    return;
  }
  int64_t msg_id = boost::lexical_cast<int64_t>(msg_content_map["id"]);
  if (msg_content_map.find("type") == msg_content_map.end() || msg_content_map["type"].empty()) {
    LOG_ERROR("CometConnection::HandleWrite --> can not find type. msg_t:" << msg_t);
    return;
  }
  int32_t msg_type = boost::lexical_cast<int32_t>(msg_content_map["type"]);
  if (msg_content_map.find("sender") == msg_content_map.end() || msg_content_map["sender"].empty()) {
    LOG_ERROR("CometConnection::HandleWrite --> can not find sender. msg_t:" << msg_t);
    return;
  }
  int64_t msg_sender = boost::lexical_cast<int64_t>(msg_content_map["sender"]);
  int64_t msg_gourp_id = 0L;
  if (msg_content_map.find("groupid") != msg_content_map.end() && msg_content_map["groupid"].empty()) {
    msg_gourp_id = boost::lexical_cast<int64_t>(msg_content_map["groupid"]);
  }
  try {
    MessageServiceAdapter::instance().setMessageDistributeStatus(uid_, msg_id, msg_type, msg_sender, msg_gourp_id,
        true);
  } catch (std::exception& e) {
    LOG_ERROR("CometConnection::HandleWrite --> error. error:" << error << " msg_t" << msg_t);
  }
}

}

